package com.zlm.app;

import com.zlm.bean.UserBehavior;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Slide;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * Author: Harbour
 * Date: 2021-05-13 22:25
 * Desc: 使用table api实现热门商品
 */
public class HotItemsWithSQLApp {
    public static void main(String[] args) throws Exception {
        // step 1 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 引入时间语义

        // step 2 读取数据源
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "Hadoop201:9092");
        properties.setProperty("group.id", "consumer-group");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");

        DataStreamSource<String> inputStream =
                env.addSource(new FlinkKafkaConsumer<String>("hot-item", new SimpleStringSchema(StandardCharsets.UTF_8), properties));

        // step 3 将数据源转换成 指定的java bean得到 data stream
        DataStream<UserBehavior> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(
                    new Long(fields[0]),
                    new Long(fields[1]),
                    new Integer(fields[2]),
                    fields[3],
                    new Long(fields[4])
            );
        });

        // step 4 创建表执行环境 使用blink版本
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // step 5 将流转换为表
        Table dataTable = tableEnv.fromDataStream(dataStream, "itemId, behavior, timestamp.rowtime as ts");

        // step 6 分组开窗
        Table aggTable = dataTable.filter("behavior = 'pv'")
                .window(Slide.over("1.hours").every("5.minutes").on("ts").as("w"))
                .groupBy("itemId, w")
                .select("itemId, w.end as windowEnd, itemId.count as cnt");

        // step 7 执行查询 获取 topN
        DataStream<Row> aggStream = tableEnv.toAppendStream(aggTable, Row.class);
        tableEnv.createTemporaryView("agg", aggStream, "itemId, windowEnd, cnt");
        tableEnv.createTemporaryView("data_view", dataStream, "itemId, behavior, timestamp.rowtime as ts");

        Table resultTable = tableEnv.sqlQuery(
             "select *" +
                "from (" +
                "   select *, ROW_NUMBER() over(partition by windowEnd order by cnt desc) as row_num" +
                "   from (" +
                "       select itemId, count(itemId) as cnt, HOP_END(ts, interval '5' minute, interval '1' hour) as windowEnd" +
                "       from data_view " +
                "       where behavior = 'pv' " +
                "       group by itemId, HOP(ts, interval '5' minute, interval '1' hour)" +
                "   )" +
                ")" +
                "where row_num <= 5"
        );
        resultTable.printSchema();
        tableEnv.toRetractStream(resultTable, Row.class).print();

        env.execute();
    }
}
